package com.mattgarner.jaddas.cli.data;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;

import com.mattgarner.jaddas.cli.net.Client;
import com.mattgarner.jaddas.cli.net.Message;
import com.mattgarner.jaddas.cli.net.Protocol;

public class BulkLoader {

	private String targetTable;
	private String inputFileName;
	private String fieldDeliminator;
	private String lineTerminator;
	private String stringEnclosure;
	private String[] selectedColumnOrder;
	
	public BulkLoader() {
		
	}
	
	public final void setTargetTable(String targetTable) {
		this.targetTable = targetTable;
	}
	
	public final void setInputFileName(String inputFileName) {
		this.inputFileName = inputFileName;
	}
	
	public final void setFieldDeliminator(String fieldDeliminator) {
		this.fieldDeliminator = fieldDeliminator;
	}
	
	public final void setLineTerminator(String lineTerminator) {
		this.lineTerminator = lineTerminator;
	}
	
	public final void setStringEnclosure(String stringEnclosure) {
		this.stringEnclosure = stringEnclosure;
	}
	
	public final void setSelectedColumnOrder(String[] selectedColumnOrder) {
		this.selectedColumnOrder = selectedColumnOrder;
	}
	
	public final void doBulkImport(Client client) throws BulkLoaderException {
		
		// validate some inputs
		if(inputFileName == null) {
			throw new BulkLoaderException("Missing Input Filename.");
		}
		if(targetTable == null) {
			throw new BulkLoaderException("Missing Target Table.");
		}
		if(fieldDeliminator == null) {
			fieldDeliminator = ";"; // default
		}
		if(lineTerminator == null) {
			lineTerminator = "\\n"; // default
		}
		if(stringEnclosure == null) {
			stringEnclosure = "\""; // default
		}
		
		int insertCount = 0;
		
		// access the file
		try {
			FileInputStream fis = new FileInputStream(inputFileName);
			BufferedReader in = new BufferedReader(new InputStreamReader(fis));
			
			/* original method of supporting arbitrary line terminators
			ByteArrayOutputStream lineBuffer = new ByteArrayOutputStream();
			int lineTerm = lineTerminator.getBytes()[0];
			while(true) {
				int nextByte = in.read();
				if(nextByte >= 0) {
					if(nextByte == lineTerm) {
						parseLine(new String(lineBuffer.toByteArray(), "UTF-8"));
					} else {
						lineBuffer.write(nextByte);
					}
				} else {
					break;
				}
			}
			*/
			
			int linesPerMessage = 10;
			byte[][] packedLinesBuffer = new byte[linesPerMessage][];
			short messageLineCursor = 0;
			
			int columnCount = 0;
			boolean fileReady = true;
			while(fileReady) {
				String line = in.readLine();
				if(line != null) {
					String[] columnStrings = parseLine(line);
					packedLinesBuffer[messageLineCursor] = packLine(messageLineCursor, columnStrings);
					if(columnCount == 0) { columnCount = columnStrings.length; }
					messageLineCursor++;
				} else {
					linesPerMessage = messageLineCursor;
					fileReady = false;
				}
				
				// send a message
				if(messageLineCursor == linesPerMessage) {
					
					ByteArrayOutputStream boasMsg = new ByteArrayOutputStream();
					DataOutputStream msgBuffer = new DataOutputStream(boasMsg);
					
					msgBuffer.writeUTF(targetTable); // target table name
					msgBuffer.writeShort(columnCount); // column count
					
					if(selectedColumnOrder != null) { // customized column order
						msgBuffer.writeShort(selectedColumnOrder.length);
						for(int c=0;c<selectedColumnOrder.length;c++) {
							msgBuffer.writeUTF(selectedColumnOrder[c]);
						}
					} else {
						msgBuffer.writeShort(0); // default column order
					}
					
					msgBuffer.writeShort(linesPerMessage); // row count
					
					for(int a=0; a<linesPerMessage; a++) {
						msgBuffer.write(packedLinesBuffer[a]);
					}
					msgBuffer.flush();
					
					client.sendMessage(new Message(Protocol.MSG_BULKLOAD_ROWS, boasMsg.toByteArray()));
					Message msgReply = client.readMessage();
					byte replyFlag = msgReply.getMessageFlag();
					
					if(replyFlag == Protocol.MSG_NAK) {
						System.out.println("<< ERR: " + msgReply.getMessageString());
						return;
					} else {
						System.out.println("<< OK: " + msgReply.getMessageString());
						insertCount += linesPerMessage;
					}
					
					// reset cursor
					messageLineCursor = 0;
				}
			}
			
			System.out.println("<< OK: Sent " + insertCount + " lines.");
			return;
			
		} catch(IOException e) {
			throw new BulkLoaderException("Input IO Error: " + e.getMessage());
		}
	}
	
	private final String[] parseLine(String line) {
		String[] columns = line.split(fieldDeliminator);
		if(stringEnclosure != null) {
			String initialReplacement = "^"+stringEnclosure;
			String terminalReplacement = stringEnclosure +"$";
			
			for(int a=0; a<columns.length; a++) {
				columns[a] = columns[a].replaceFirst(initialReplacement, "").replaceFirst(terminalReplacement, "");
			}
		}
		return columns;
	}
	
	private final byte[] packLine(short msgLineNumber, String[] columns) throws BulkLoaderException {
		
		ByteArrayOutputStream boasLine = new ByteArrayOutputStream();
		DataOutputStream lineBuffer = new DataOutputStream(boasLine);
		
		try {
			lineBuffer.writeShort(msgLineNumber);
			for(int a=0; a<columns.length; a++) {
				lineBuffer.writeUTF(columns[a]);
			}
			lineBuffer.flush();
			return boasLine.toByteArray();
		} catch(Exception e) {
			throw new BulkLoaderException("packLine: " + e.getMessage());
		}
	}
	

}
